package com.tc.zookeeper.curator;

import com.alibaba.fastjson.JSONObject;
/*import com.vip.hermes.common.model.event.EventDetailData;
import com.vip.hermes.core.util.HermesUtil;
import com.vip.hermes.event.reporter.EventSlf4jReporter;*/
import lombok.Setter;
import org.apache.commons.lang3.StringUtils;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.imps.CuratorFrameworkImpl;
import org.apache.curator.framework.recipes.cache.ChildData;
import org.apache.curator.framework.recipes.cache.PathChildrenCache.StartMode;
import org.apache.curator.framework.state.ConnectionState;
import org.apache.curator.framework.state.ConnectionStateListener;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.net.InetAddress;
import java.net.UnknownHostException;
import java.util.Iterator;
import java.util.List;
import java.util.Map.Entry;
import java.util.UUID;
import java.util.concurrent.Executor;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicBoolean;

public class EnhancedCuratorFrameworkImpl extends CuratorFrameworkImpl implements EnhancedCuratorFramework {

	private static final Logger LOGGER = LoggerFactory.getLogger(EnhancedCuratorFrameworkImpl.class);

	@Setter
	private String targetConnectString;

	private String uniqueId;

	private AtomicBoolean startUpdateFlag = new AtomicBoolean(false);

	private Executor executor = Executors.newCachedThreadPool();

	public EnhancedCuratorFrameworkImpl(EnhancedCuratorFrameworkFactory.Builder builder) {
		super(builder.toCuratorBuilder());

		this.uniqueId = String.valueOf(UUID.randomUUID().getLeastSignificantBits());
	}

	@Override
	public void start() {
		super.start();

		if (StringUtils.isBlank(targetConnectString)) {
			return;
		}

		this.getConnectionStateListenable().addListener(new ConnectionStateListener() {
			public void stateChanged(CuratorFramework client, ConnectionState newState) {
				stateChangedProcess(newState);
			}
		});
	}

	public void stateChangedProcess(ConnectionState newState) {
		switch (newState) {
		case LOST:
			startUpdateFlag.set(true);
			lostConnection(targetConnectString, uniqueId);
			break;
		case RECONNECTED:
			startUpdateFlag.set(false);
			reconnected(targetConnectString, uniqueId);
			break;
		default:
			break;
		}
	}

	private static void reconnected(final String connectString, final String uniqueId) {
		LOGGER.info(">>>>>>>>reconnect, connectString:" + connectString);
		reportEvent(connectString, "RECONNECTED", uniqueId);
	}

	private static void lostConnection(String connectString, final String uniqueId) {
		LOGGER.warn(">>>>>>>>lostConnection, connectString:" + connectString);
		reportEvent(connectString, "LOST", uniqueId);
	}

	private static void reportEvent(String targetConnectString, String connectionState, String uniqueId) {
		LOGGER.info("start reportEvent....");
		try {
	/*		JSONObject json = new JSONObject();
			String localAddress = HermesUtil.localIp4Str();
			json.put("ipAddress", localAddress);
			String localHostName = getLocalHostName();
			json.put("hostName", localHostName);
			json.put("targetConnectString", targetConnectString);
			json.put("uniqueId", uniqueId);

			String[] tags = new String[3];
			tags[0] = localAddress;
			tags[1] = localHostName;
			tags[2] = targetConnectString;

			EventSlf4jReporter reporter = new EventSlf4jReporter();
			EventDetailData detail = new EventDetailData();
			detail.setMessage(json.toJSONString());
			detail.setName("ZKCLIENT.CONNECT." + connectionState);
			detail.setTitle("ZKCLIENT.CONNECT." + connectionState);
			detail.setType("ZKCLIENT.NETWORK.EXCEPTION");
			detail.setLevel("CRITICAL");
			detail.setTag(tags);
			reporter.report(detail);*/
		} catch (Throwable e) {
			LOGGER.warn("reportEvent error :" + e.getMessage());
		}
	}

	private static String getLocalHostName() {
		try {
			return InetAddress.getLocalHost().getHostName();
		} catch (UnknownHostException ignore) {
		}
		return null;
	}

	@Override
	public byte[] watchNode(String path, EnhancedNodeCacheListener listener) throws Exception {
		byte[] result = null;
		EnhancedNodeCache cache = null;
		try {
			if (this.checkExists().forPath(path) != null) {
				cache = new EnhancedNodeCache(this, path);
				cache.getListenable().addListener(listener, executor);

				cache.start(true);

				cache.getListenable().addListener(new MetricNodeCacheListener(cache, targetConnectString), executor);

				result = cache.getCurrentData().getData();
			}
		} catch (Exception e) {
			LOGGER.error("watchNode error", e);
			reportEvent(targetConnectString, "RETRYFAILED", uniqueId);

			try {
				if (cache != null) {
					cache.close();
				}
			} catch (IOException e1) {
				LOGGER.error("close cache error:" + e1.getMessage());
			}
			throw e;
		}
		return result;
	}

	@Override
	public List<ChildData> watchChildrenNodes(String path, EnhancedPathChildrenCacheListener listener) throws Exception {
		List<ChildData> list = null;
		EnhancedPathChildrenCache cache = null;
		try {
			if (this.checkExists().forPath(path) != null) {
				cache = new EnhancedPathChildrenCache(this, path, true);
				cache.getListenable().addListener(listener, executor);
				cache.getListenable().addListener(new MetricPathChildrenCacheListener(path, targetConnectString),
						executor);
				cache.start(StartMode.BUILD_INITIAL_CACHE);

				list = cache.getCurrentData();
			}
		} catch (Exception e) {
			LOGGER.error("watchChrildrenNodes error", e);
			reportEvent(targetConnectString, "RETRYFAILED", uniqueId);

			try {
				if (cache != null) {
					cache.close();
				}
			} catch (IOException e1) {
				LOGGER.error("close cache error:" + e1.getMessage());
			}
			throw e;
		}
		return list;
	}

	@Override
	public void watchChildrenAddRemove(String path, EnhancedChildrenAddRemoveListener listener) throws Exception {
		EnhancedPathChildrenCache cache = null;
		try {
			if (this.checkExists().forPath(path) != null) {
				cache = new EnhancedPathChildrenCache(this, path, true);
				cache.getListenable().addListener(listener, executor);
				cache.getListenable().addListener(new MetricPathChildrenCacheListener(path, targetConnectString),
						executor);
				cache.start(StartMode.BUILD_INITIAL_CACHE);
			}
		} catch (Exception e) {
			LOGGER.error("watchChildrenAddRemove error", e);
			reportEvent(targetConnectString, "RETRYFAILED", uniqueId);

			try {
				if (cache != null) {
					cache.close();
				}
			} catch (IOException e1) {
				LOGGER.error("close cache error:" + e1.getMessage());
			}
			throw e;
		}
	}

	@Override
	public void close() {
		try {
			Iterator<Entry<String, EnhancedCuratorFramework>> iterator = EnhancedCuratorFrameworkFactory
					.getCachedCuratorFrameworkMap().entrySet().iterator();
			while (iterator.hasNext()) {
				if (iterator.next().getKey().equals(this.targetConnectString)) {
					iterator.remove();
				}
			}
		} catch (Exception e) {
			LOGGER.error("close  EnhancedCuratorFramework error:" + e.getMessage());
		}

		super.close();
	}

}
